
#include "scheduler.h"
#include "common.h"
#include "config.h"
#include "undertopomgr.h"
#include "eventdefs.h"
#include "simulator.h"
#include "peerloger.h"
#include "node.h"
#include "chordnode.h"
#include "chordgod.h"
#include "randomrangepro.h"
#include "chordkeyinfo.h"
#include "datasource.h"


Scheduler* Scheduler::instance_ = NULL;

Scheduler::Scheduler()
	: Object()
	, disaplay_intv_(20.0)
{
	instance_ = this;
	NEW_OBJECT( DisplayTimer, disp_timer_,  (this) );
}

Scheduler::~Scheduler()
{
	RELEASE_OBJ( disp_timer_ );
}

bool Scheduler::create()
{
	NEW_OBJECT( Scheduler, instance_, () );
	return true;
}

void Scheduler::destroy()
{
	if ( instance_!=NULL ){
		RELEASE_OBJ( instance_ )
	}
}

bool Scheduler::scheduleAll()
{
	bool ret = scheduleNodes();

	if ( config.scheduler_.search_.start_time >0 )
	{
		NEW_VAR_OBJECT2( Event, CmdEvent, event, (this, CmdEvent::CMD_SCH_SEARCH) );
		event->setTime( config.scheduler_.search_.start_time );
		Simulator::instance().scheduleEvent( event );
		RELEASE_OBJ( event );
	}


	if ( config.scheduler_.dump_time>0 )
	{//dump the chord net to file after all node online
		NEW_VAR_OBJECT2( Event, CmdEvent, event, (this, CmdEvent::CMD_DUMP_CHORD_NET) );
		sim_time_t d_time = config.scheduler_.dump_time;
		event->setTime( d_time );
		Simulator::instance().scheduleEvent( event );
		RELEASE_OBJ( event );
		PeerLoger::instance().logScheduler( 0, "chordnet will be dumped at %f", d_time );
	}

	if ( config.scheduler_.merge_.start_time>0 ){
		NEW_VAR_OBJECT2( Event, CmdEvent, event, (this, CmdEvent::CMD_MERGE) );
		sim_time_t m_time = config.scheduler_.merge_.start_time;
		event->setTime( m_time );
		Simulator::instance().scheduleEvent( event );
		RELEASE_OBJ( event );
		PeerLoger::instance().logScheduler( 0, "chordnet will be dumped at %f", m_time );
	}

	disp_timer_->resched( disaplay_intv_ );

	return ret;
}

bool Scheduler::scheduleNodes()
{
	bool ret = false;
	switch( config.scheduler_.node_source ){
		case Config::SCH_SEQUENTIAL:
			ret = scheduleNodesSequential();
			break;
		case Config::SCH_RANDOM:
			ret = scheduleNodesRandom();
			break;
		case Config::SCH_TRACE:
			ret = scheduleNodesFromTrace( config.scheduler_.trace_file.c_str() );
			break;
		default:
			assert(false);
	}

	return ret;
}

bool Scheduler::scheduleNodesSequential()
{
	uint32_t n_size = UnderTopoMgr::instance().networkSize();
	sim_time_t int_arrv_t = config.scheduler_.interval;
	sim_time_t this_arrival_time = 0.0;
	uint32_t i = 0;

	if (config.chordgod_.initial_chord_size > 0) return true;

	for ( ; i<n_size; i++ ){
		//generate peer arrival and departure events
		NEW_VAR_OBJECT2( Event, NodeJoinEvent, event, (this, i) );
		event->setTime( this_arrival_time );
		Simulator::instance().scheduleEvent( event );
		RELEASE_OBJ( event );
		this_arrival_time += int_arrv_t;
	}

	uint32_t online_cnt = i;
	sim_time_t last_arrival_time = this_arrival_time;

	sim_time_t failure_time = config.scheduler_.failure_time;
	double     failure_portion = config.scheduler_.failure_portion;
	if ( failure_time>0 && failure_portion>0 ){
		RandomRangePro    rr(0,(uint32_t)online_cnt, (uint32_t)(n_size*failure_portion));
		//RandomRangePro    rr(0,(uint32_t)online_cnt, 3);
		while( rr.hasNext() ){ 
			i = rr.next();
			NEW_VAR_OBJECT2( Event, NodeDepartureEvent, event, (this,i) );
			event->setTime( failure_time );
			Simulator::instance().scheduleEvent( event );
			RELEASE_OBJ( event );
		}
	}
#if 0
	{//dump the chord net to file after all node online
		NEW_VAR_OBJECT2( Event, CmdEvent, event, (this, CmdEvent::CMD_DUMP_CHORD_NET) );
		sim_time_t d_time = 450.0;
		event->setTime( d_time );
		Simulator::instance().scheduleEvent( event );
		RELEASE_OBJ( event );
		PeerLoger::instance().logScheduler( 0, "chordnet will be dumped at %f", d_time );
	}
#endif

	PeerLoger::instance().logScheduler( 0, "%d nodes scheduled, last_arrival_time:%f", online_cnt, last_arrival_time );
	return true;
}

bool Scheduler::scheduleNodesRandom()
{
	uint32_t n_size = UnderTopoMgr::instance().networkSize();
	sim_time_t interval = config.scheduler_.interval;
	sim_time_t lifespan = config.scheduler_.lifespan;
	sim_time_t this_arrival_time = 0.0;

	uint32_t i = 0;
	for ( ; i<config.scheduler_.init_size; i++ ){
		NEW_VAR_OBJECT2( Event, NodeJoinEvent, event, (this, i) );
		event->setTime( 0.0 );
		Simulator::instance().scheduleEvent( event );
		RELEASE_OBJ( event );
	}

	this_arrival_time = config.scheduler_.start_time;

	if ( config.scheduler_.start_time>=0 ){
		for ( ; i<n_size; i++ ){
			//generate peer arrival and departure events
			NEW_VAR_OBJECT2( Event, NodeJoinEvent, event_on, (this, i) );
			event_on->setTime( this_arrival_time );
			Simulator::instance().scheduleEvent( event_on );
			RELEASE_OBJ( event_on );

			sim_time_t duration = rng.expntl(lifespan);
			sim_time_t failure_time = this_arrival_time + duration;
			NEW_VAR_OBJECT2( Event, NodeDepartureEvent, event_off, (this,i) );
			event_off->setTime( failure_time );
			Simulator::instance().scheduleEvent( event_off );
			RELEASE_OBJ( event_off );

			PeerLoger::instance().logScheduler(0, "node:%u arrv_time:%f duration:%f off_time:%f", i, this_arrival_time, duration, failure_time);
			this_arrival_time += rng.expntl(interval);
		}
	}

	sim_time_t last_arrival_time = this_arrival_time;
	uint32_t online_cnt = i;
	PeerLoger::instance().logScheduler( 0, "%u nodes scheduled, last_arrival_time:%f", online_cnt, last_arrival_time );
	fprintf( stderr, "\n%u nodes scheduled, last_arrival_time:%f\n", online_cnt, last_arrival_time );

	return true;
}

bool Scheduler::scheduleNodesFromTrace(const char *filename)
{
	DataSource ds;
}

void Scheduler::onEvent( Event *event )
{
	if ( event->getType() ==  Event::NODE_JOIN ){
		NodeJoinEvent* join_event = static_cast<NodeJoinEvent*>(event);
		UnderTopoMgr::instance().node(join_event->nodeID())->onJoining();
	} else if ( event->getType() ==  Event::NODE_DEPARTURE ){
		NodeDepartureEvent* dept_event = static_cast<NodeDepartureEvent*>(event);
		UnderTopoMgr::instance().node(dept_event->nodeID())->onDeparture();
	} else if ( event->getType() == Event::COMMAND ){
		CmdEvent* cmd_event = static_cast<CmdEvent*>(event);
		executeCmd( cmd_event );
	} else{
			assert(false);
	}
}

void Scheduler::onTimer( Timer *timer )
{
	assert(false);
}

void Scheduler::executeCmd( CmdEvent *cmd_event )
{
	switch( cmd_event->cmd() ){
		case CmdEvent::CMD_DUMP_CHORD_NET:
			ChordGod::instance().dumpChordNet();
			break;
		case CmdEvent::CMD_SCH_SEARCH:
			scheduleSearchEvent();
			break;
		case CmdEvent::CMD_MERGE:
			ChordGod::instance().startMerge();
			break;
		default:
			assert(false);
	}
}

void Scheduler::displayProgress()
{
	sim_time_t now = Simulator::instance().clock();
	fprintf(stderr, "\rcur_time:%f            ", now);
	disp_timer_->resched( disaplay_intv_ );
}

void Scheduler::scheduleSearchEvent()
{
	switch( config.scheduler_.search_.source ){
		case Config::SCH_SEQUENTIAL:
			scheduleSearchSequential();
			break;
		//case Config::SCH_MERGE:
			//scheduleSearchMerge();
			//break;
		case Config::SCH_RANDOM:
		case Config::SCH_TRACE:
		default:
			assert(false);
			break;
	}
}

void Scheduler::scheduleSearchMerge()
{
	//assuming two nets, 0 and 1, are merging to net 2
	chord_net_id_t new_net_id = 2;
	chord_net_id_t cur_net_id = 0;
	chord_net_id_t other_net_id = 1;

	int sched_cnt = 0;

	sim_time_t     this_arrival_time = config.scheduler_.search_.start_time;
	int 		   search_number = config.scheduler_.search_.number;

	//for each nets, generate config.
	for ( ; cur_net_id<1; cur_net_id++ ){//only support two nets now
		assert( new_net_id==2 );
		if ( cur_net_id==0 ) other_net_id = 1;
		else other_net_id = 0;
		ChordNetInfo&  cur_net_info = ChordGod::instance().getNetInfo( cur_net_id );
		ChordNetInfo&  other_net_info = ChordGod::instance().getNetInfo( other_net_id );
		for ( int i=0; i<search_number; i++ ){
			uint32_t index = (uint32_t)i;
			chord_key_t key = other_net_info.key_tuples_[index][0];
			ChordNode* n = ChordGod::instance().getNodeInChordNet( cur_net_id, key );
			if ( NULL==n ) break;
			NEW_VAR_OBJECT2( Event, SearchEvent, event, (n, new_net_id, key) );
			event->setTime( this_arrival_time );
			Simulator::instance().scheduleEvent( event );
			this_arrival_time += config.scheduler_.search_.interval;
			sched_cnt ++;
			RELEASE_OBJ( event );
		}
	}
	fprintf( stderr, "\nscheduleSearchMerge count:%d last_arrival_time:%f\n", sched_cnt, this_arrival_time );
}

void Scheduler::scheduleSearchSequential()
{
	//generate sequential keys for test
#if 0
	std::vector<chord_key_t> random_list;
	int network_size = (int)UnderTopoMgr::instance().networkSize();
	for( int count=0; count<network_size; count++ ){
		random_list.push_back(count);
	}
	std::random_shuffle( random_list.begin(), random_list.end() );
#endif

	if ( config.scheduler_.merge_.start_time>0 )
		return scheduleSearchMerge();

	int i = 0;
	chord_net_id_t net_id = 0;
	chord_key_t    key    = 0;
	sim_time_t     this_arrival_time = config.scheduler_.search_.start_time;
	ChordNetInfo&  net_info = ChordGod::instance().getNetInfo( net_id );
	uint32_t	   key_no = (net_info.number_of_keys_);
	int 		   search_number = config.scheduler_.search_.number;
	//RandomRangePro    rr(0,(uint32_t)key_no, (uint32_t)search_number);

	if ( search_number>(int)key_no ) {
		search_number = key_no;
		fprintf( stderr, "Warning: search number is greater than the total key number, search number is reduced to %d\n", search_number );
	}

	for( ; i<search_number; i++ ){
		//uint32_t index = rr.next();
		uint32_t index = i;
		key = net_info.key_tuples_[index][0];
		ChordNode* n = ChordGod::instance().getNodeInChordNet( net_id, key );
		if ( NULL==n ) break;
		NEW_VAR_OBJECT2( Event, SearchEvent, event, (n, net_id, key) );
		event->setTime( this_arrival_time );
		Simulator::instance().scheduleEvent( event );
		this_arrival_time += config.scheduler_.search_.interval;
		RELEASE_OBJ( event );
	}

	fprintf( stderr, "\nscheduleSearchSequential count:%d\n", i );
}
